package net.self.kafka.consumer;

import org.apache.ibatis.session.SqlSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;

/**
 * The partition offset are stored in an external storage. In this case in a file system.
 * <p/>
 */
public class OffsetManager {

    private final Logger logger = LoggerFactory.getLogger(OffsetManager.class);

    /**
     * Overwrite the offset for the topic in an external storage.
     *
     * @param topic     - Topic name.
     * @param partition - Partition of the topic.
     * @param offset    - offset to be stored.
     */
    public void saveOffsetInExternalStore(String topic, int partition, long offset, Transaction jedis) {

        try {
            jedis.set(storageName(topic, partition),String.valueOf(offset));
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    public void saveOffsetInExternalStore(String topic, int partition, long offset, SqlSession sqlSession){
        try {
            //todo : insert into db offset
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * @return the last offset + 1 for the provided topic and partition.
     */
    long readOffsetFromExternalStore(String topic, int partition, Jedis jedis) {

        try {
            String key = jedis.get(storageName(topic,partition));
            if (key == null){
                logger.info("can't find the offset with this topic&partition,reset to 0");
                return 0;
            }
            return Long.parseLong(jedis.get(storageName(topic, partition)).toString()) + 1;
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

//    long readOffsetFromExternalStore(String topic, int partition, SqlSession sqlSession) {
//
//        try {
//            OffsetMapper offsetMapper = new OffsetMapper(sqlSession);
//            return offsetMapper.getOffset(storageName(topic, partition)) + 1;
//        } catch (Exception e) {
//            throw new RuntimeException(e);
//        }
//    }

    private String storageName(String topic, int partition) {
        return topic + "-" + partition;
    }

}

